# -*- coding: utf-8 -*-
from typing import Iterable, Dict, List
from airflow import DAG
from airflow.models import Variable, SlaMiss, TaskInstance
from utils.hooks.phone_call_hook import PhoneCallHook


def call_phone_by_email(conn_id='phone_call',
                        alert_name: str = "任务失败",
                        voice_name: str = "任务失败",
                        project: str = "大数据", ):
    def _call_phone_by_email(context):
        task = context['task']
        users = _get_ldap_phone_from_tasks([task])
        res = PhoneCallHook(conn_id=conn_id).run(voice_name=voice_name, alert_name=alert_name, users=users, project=project)
        print('result:\n', f'reason: {res.reason}, status_code: {res.status_code}, text: {res.text}')

    return _call_phone_by_email


def call_phone_on_sla_miss(conn_id='phone_call',
                           alert_name: str = "任务超时",
                           voice_name: str = "任务超时",
                           project: str = "大数据", ):
    def _call_phone_on_sla_miss(dag: DAG, task_list: [str], blocking_task_list: [str],
                                slas: [SlaMiss], blocking_tis: [TaskInstance]):
        tasks = [dag.get_task(sla.task_id) for sla in slas]
        users = _get_ldap_phone_from_tasks(tasks)
        res = PhoneCallHook(conn_id=conn_id).run(users=users, alert_name=alert_name, voice_name=voice_name, project=project)
        print('result:\n', f'reason: {res.reason}, status_code: {res.status_code}, text: {res.text}')

    return _call_phone_on_sla_miss

import datetime
import json
import pymysql

def fetchMonitorTasks():
    try:
        dt = (datetime.datetime.now() + datetime.timedelta(days=-1)).strftime("%Y-%m-%d")
        airflow_customize_config = json.loads(Variable.get(key='airflow_customize_config', default_var=''))
        db = pymysql.connect(host=airflow_customize_config.get("host"),
                             user=airflow_customize_config.get("username"),
                             password=airflow_customize_config.get("password"),
                             port=3306,# 端口
                             database=airflow_customize_config.get("db"),
                             charset='utf8')
        cursor = db.cursor()
        sql = "select json_concat_info from email_and_telphone where dt= '{}'  limit 1 ".format(dt)
        cursor.execute(sql)
        res = cursor.fetchall()[0][0]
    except Exception as err:
        print(err)
        return Variable.get('phone_map', deserialize_json=True, default_var={})
    else:
        return eval(res)

def checkIfDaytime():
    currentHour = datetime.datetime.now().strftime('%H')
    airflow_calltime_config = json.loads(Variable.get(key='airflow_calltime_config',
                                                      default_var='{"startHour": 10, "endHour": 22}'))
    startHour = airflow_calltime_config.get("startHour")
    endHour = airflow_calltime_config.get("endHour")
    if (int(startHour) <= int(currentHour) <= int(endHour)):
        print("current Hour is {}, which between {} and {}, so telphone alert function will be close !".format(currentHour, startHour, endHour))
        return 1
    else:
        print("current Hour is {}, which out of {} and {}, so telphone alert function will be work !".format(currentHour, startHour, endHour))
        return 0

def _get_ldap_phone_from_tasks(tasks: Iterable) -> List[Dict[str, str]]:
    # phone_map = Variable.get('phone_map', deserialize_json=True, default_var={})
    # if checkIfDaytime():
    #     phone_map = '{"yl_etl@yl-scm.com": [{"ldap": "yl_etl", "phone": ""}], "yl_bigdata@yl-scm.com": [{"ldap": "yl_bigdata", "phone": ""}}'
    # else:
    #     phone_map = fetchMonitorTasks()
    phone_map = fetchMonitorTasks()
    #phone_map = '{"yl_etl@yl-scm.com": [{"ldap": "yl_etl", "phone": ""}], "yl_bigdata@yl-scm.com": [{"ldap": "yl_bigdata", "phone": ""}}'
    emails = set(e.strip()
                 for t in tasks if t.email is not None
                 for e in (t.email.split(',') if isinstance(t.email, str) else t.email))
    return [{'username': user['ldap'], 'phone': user['phone']}
            for e in emails if e in phone_map
            for user in phone_map[e]]

